package com.atbeijing.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class ConsumerManual {
    public static void main(String[] args) {
        Properties conf = new Properties();
        try {
            conf.load(ConsumerManual.class.getClassLoader().getResourceAsStream("consumer.properties"));
        } catch (IOException e) {
            e.printStackTrace();
        }
        conf.setProperty("group.id", "test03");
        conf.setProperty("enable.auto.commit", "false");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(conf);

        //消费指定topic
        consumer.subscribe(Collections.singleton("first"));
        //循环拉取
        while (true) {
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(10));
            //消费消息
            for (ConsumerRecord<String, String> consumerRecord : poll) {
                System.out.println(consumerRecord);
            }
            //手动同步提交offset,报告消费到哪里了,当消费的动作和offset提交的动作在一个事务时能实现精准一次性消费
            consumer.commitSync();
        }

        //关闭consumer
//            consumer.close();
    }
}
